package io.kiki.sba.registry.client.remoting;


import io.kiki.sba.registry.client.api.ClientConfig;
import io.kiki.sba.registry.client.api.Configurator;
import io.kiki.sba.registry.client.api.Publisher;
import io.kiki.sba.registry.client.api.Subscriber;
import io.kiki.sba.registry.client.provider.RegisterCache;
import io.kiki.sba.registry.client.task.TaskEvent;
import io.kiki.sba.registry.client.task.Worker;
import io.kiki.stack.netty.channelManager.Channel;
import io.kiki.stack.netty.channelManager.ChannelEventProcessor;
import io.kiki.stack.netty.channelManager.ChannelEventType;
import io.kiki.stack.netty.exception.RemotingException;
import io.kiki.stack.netty.userProcessor.UserProcessor;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.Map.Entry;

@Setter
@Getter
public class ClientChannel implements Client {

    private static final Logger logger = LoggerFactory.getLogger(ClientChannel.class);
    /**
     * The Reconnecting delay.
     */
    private static final int RECONNECTING_DELAY = 5000;

    private io.kiki.stack.netty.Client client;
    private ServerManager serverManager;
    private List<UserProcessor> userProcessorList;
    private Map<ChannelEventType, ChannelEventProcessor> connectionEventProcessorMap;
    private ClientConfig clientConfig;
    private Channel channel;
    private RegisterCache registerCache;
    private Worker worker;


    public ClientChannel(ServerManager serverManager, List<UserProcessor> userProcessorList, Map<ChannelEventType, ChannelEventProcessor> connectionEventProcessorMap, RegisterCache registerCache, ClientConfig clientConfig) {
        this.client = new io.kiki.stack.netty.Client();
        this.serverManager = serverManager;
        this.userProcessorList = userProcessorList;
        this.connectionEventProcessorMap = connectionEventProcessorMap;
        this.registerCache = registerCache;
        this.clientConfig = clientConfig;
    }


    @Override
    public void init() {
        for (UserProcessor userProcessor : userProcessorList) {
            client.registerUserProcessor(userProcessor);
        }

        if (null != connectionEventProcessorMap) {
            for (Entry<ChannelEventType, ChannelEventProcessor> entry : connectionEventProcessorMap.entrySet()) {
                client.addChannelEventProcessor(entry.getKey(), entry.getValue());
            }
        }

        client.init();
    }


    @Override
    public void ensureConnected() throws InterruptedException {
        if (isConnected()) {
            return;
        }
        while (!connect()) {
            Thread.sleep(ClientChannel.RECONNECTING_DELAY);
        }
    }


    private boolean connect() {
        Random random = new Random();
        Channel channel = null;
        List<Server> list = new ArrayList<Server>(serverManager.getServerList());
        // shuffle server list to make server connections as discrete as possible
        Collections.shuffle(list);
        int choosed = 0;
        for (int i = 0; i < list.size(); i++) {
            try {
                // Power of Two Choices
                if (list.size() > 1) {
                    choosed = list.get(i).getWeight() > list.get((i + 1) % list.size()).getWeight() ? (i + 1) % list.size() : i;
                }
                channel = connect(list.get(choosed));
                if (null != channel && channel.isFine()) {
                    resetRegister();
                    logger.info("[Connect] Successfully connected to server: {}", list.get(choosed));
                    break;
                } else {
                    recycle(channel);
                }

                Thread.sleep(random.nextInt(RECONNECTING_DELAY));
            } catch (Exception e) {
                logger.error("[Connect] Failed trying connect to {}", list.get(choosed), e);
            }
        }

        if (null != channel && channel.isFine()) {
            this.channel = channel;
            return true;
        }
        return false;
    }


    @Override
    public Object invokeSync(Object request) throws InterruptedException, RemotingException {
        if (!isConnected()) {
            throw new IllegalStateException("Not connected");
        }

        return client.invokeSync(channel, request, clientConfig.getInvokeTimeout());
    }

    private void recycle(Channel connection) {
        if (null == connection) {
            return;
        }

        client.closeChannel(connection.getUrl());
    }

    private Channel connect(Server server) {
        try {
            return client.getChannel(server.getUrl(), clientConfig.getConnectTimeout());
        } catch (Exception e) {
            logger.error("[connection] Create connection error, {}", server, e);
            return null;
        }
    }

    private void resetRegister() {
        try {
            List<TaskEvent> list = new ArrayList<TaskEvent>();

            List<Publisher> publishers = registerCache.getPublisherList();
            for (Publisher publisher : publishers) {
                try {
                    publisher.reset();
                    list.add(new TaskEvent(publisher));
                } catch (Exception e) {
                    logger.error("[connection] Publisher reset error, {}", publisher, e);
                }
            }

            List<Subscriber> subscriberList = registerCache.getSubscriberList();
            for (Subscriber subscriber : subscriberList) {
                try {
                    subscriber.reset();
                    list.add(new TaskEvent(subscriber));
                } catch (Exception e) {
                    logger.error("[connection] Subscriber reset error, {}", subscriber, e);
                }
            }

            List<Configurator> configuratorList = registerCache.getConfiguratorList();
            for (Configurator configurator : configuratorList) {
                try {
                    configurator.reset();
                    list.add(new TaskEvent(configurator));
                } catch (Exception e) {
                    logger.error("[connection] Configurator reset error, {}", configurator, e);
                }
            }

            worker.schedule(list);
            logger.info("[reset] {} publishers and {} subscribers has been reset", publishers.size(), subscriberList.size());
        } catch (Exception e) {
            logger.error("[reset] Reset register after reconnect error", e);
        }
    }


    public String getRemoteAddress() {
        if (null != channel) {
            return channel.getRemoteIp();
        }
        return null;
    }


    @Override
    public boolean isConnected() {
        return channel != null && channel.isFine();
    }


    public void destroy() {
        if (null != channel) {
            channel.close();
        }
        if (null != client) {
            client.shutdown();
        }
    }


}
